package com.atguigu.realtime.app.dws;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.bean.KeywordStats;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.function.IkAnalyzer;
import com.atguigu.realtime.function.KwProduct;
import com.atguigu.realtime.util.FlinkSinkUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2022/2/19 14:19
 */
public class DwsProductKeywordApp extends BaseSQLApp {
    public static void main(String[] args) {
        new DwsProductKeywordApp().init(4005, 1, "DwsProductKeywordApp");
    }
    
    @Override
    protected void run(StreamTableEnvironment tEnv) {
        //1. 消费指定的topic
        tEnv.executeSql("create table dws_product(" +
                            "   stt string, " +
                            "   edt string, " +
                            "   sku_name string," +
                            "   click_ct bigint, " +
                            "   order_ct bigint," +
                            "   cart_ct bigint" +
                            ")with(" +
                            "   'connector' = 'kafka', " +
                            "   'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "', " +
                            "   'properties.group.id' = 'DwsProductKeywordApp', " +
                            "   'topic' = '" + Constant.TOPIC_DWS_PRODUCT + "', " +
                            "   'scan.startup.mode' = 'latest-offset', " +
                            "   'format' = 'json'" +
                            ")");
        
        // 2. 过滤出来三个count至少把一个不为0
        Table t1 = tEnv.sqlQuery("select " +
                                     "* " +
                                     "from dws_product " +
                                     "where click_ct > 0 " +
                                     "or order_ct > 0 " +
                                     "or cart_ct > 0");
        tEnv.createTemporaryView("t1", t1);
        
        // 3 对sku_name进行分词
        tEnv.createTemporaryFunction("ik_analyzer", IkAnalyzer.class);
        Table t2 = tEnv.sqlQuery("select " +
                                     " stt, " +
                                     " edt, " +
                                     " word, " +
                                     " click_ct," +
                                     " order_ct," +
                                     " cart_ct " +
                                     "from t1 " +
                                     ", lateral table(ik_analyzer(sku_name))");
        tEnv.createTemporaryView("t2", t2);
        // 4. 自定义table函数, 把三个count变成3行
        tEnv.createTemporaryFunction("kw_product", KwProduct.class);
        Table t3 = tEnv.sqlQuery("select" +
                                     " stt," +
                                     " edt," +
                                     " word keyword," +
                                     " source," +
                                     " ct " +
                                     "from t2 " +
                                     "join lateral table(kw_product(click_ct, order_ct, cart_ct)) on true");
        
        tEnv.createTemporaryView("t3", t3);
        
        Table result = tEnv.sqlQuery("select" +
                                         " stt, edt, keyword, source," +
                                         " sum(ct) ct," +
                                         " unix_timestamp()*1000 as ts  " +
                                         "from t3 " +
                                         "group by stt, edt, keyword, source");
        // 5. 数据写入到ClickHouse中
        tEnv
            .toRetractStream(result, KeywordStats.class)
            .filter(t -> t.f0)
            .map(t -> t.f1)
            .addSink(FlinkSinkUtil.getClickHouseSink("gmall2022", "keyword_stats_2022", KeywordStats.class));
        
    }
}
/*
dws 商品主题宽表
窗口    sku_name     click_ct     order_ct        cart_ct
0-5     小米手机       30            10             50
0-5     华为手机       30            10             50
-------
使用分词自定义函数达到这个效果  其实已经完成
0-5     小米       30            10             50
0-5     手机       30            10             50

列转行: 制表函数  传递三个count, 变成3行两列

-----
最终数据: 写到关键词主题宽表的时候:

0-5   小米     "click"   30
0-5   小米     "order"   10
0-5   小米     "cart"    50
0-5   手机     "click"   30
0-5   手机     "order"   10
0-5   手机     "cart"    50

0-5   手机     "click"   30
0-5   手机     "order"   10
0-5   手机     "cart"    50



 */